package com.teamsun.mqtt.service;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ExecutorService;
import java.util.stream.Stream;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;

import com.teamsun.mqtt.entity.SendMsg;
import com.teamsun.mqtt.entity.SendMsgMapper;
import com.teamsun.mqtt.entity.vo.DelayOCalQueue;
import com.teamsun.mqtt.entity.vo.DelaySnedMsg;
import com.teamsun.mqtt.entity.vo.DelayWillRemoveMsg;
import com.teamsun.mqtt.entity.vo.SendableMsg;

import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.util.ReferenceCountUtil;

/**
 * message 实际操作类
 * 
 * @author tzj
 * 
 */
public class MessagePushService {

	public  final int MSG_SENDSUCCESS = 2;
	public  final int MSG_SENDUNSUCCESS = 1;
	public  final int MSG_SENDERROR = 0;

	public   boolean usedb = false;
	
	Logger logger=Logger.getLogger(getClass());

	 ConcurrentHashMap<String, Channel> str2channel ;
	/**
	 * 每一个客户端订阅的主题
	 */
	 ConcurrentNavigableMap<String, BlockingQueue<String>> submap;
	/**
	 * 未成功发送的队列
	 */
	 @Autowired
	 DelayOCalQueue delayOCalQueue;
	/**
	 * 成功发送队列  超过长度就存储在数据库
	 */
	 BlockingQueue<SendMsg> readUpdateSendedQue;

	 ExecutorService dboptgroup;

	 @Autowired
	 ReSendService reSendService;
	 
	 @Autowired
	 SendMsgMapper sendMsgMapper;
	 
	 private  long  sendtimes;

	/**
	 * 对客户端发送数据 获取消息的主题 根据主题得到客户端 把消息发送给这些客户端
	 * 
	 * @param sendableMsg 要发送的对象
	 */
	
	public  void sendPubMsg(SendableMsg sendableMsg) {

		
		Iterator<Entry<String, BlockingQueue<String>>> iterator=
				submap.tailMap(sendableMsg.getTopname(),true).entrySet().iterator();
	
		sendableMsg.getMsgContent().retain();
		while(iterator.hasNext()){
			
			Entry<String, BlockingQueue<String>> entry=iterator.next(); 
			String key=entry.getKey();
			
			if(!key.startsWith(sendableMsg.getTopname()))
				continue;
			
			BlockingQueue<String> idens= entry.getValue();
			if (idens == null||idens.isEmpty())
				continue;
			
			Stream<String> stream=idens.parallelStream();
			stream.forEach((String clientid)->{
					sendMsg2Client(new SendableMsg(sendableMsg).setRecvclientid(clientid));
					sendtimes++;
					if (sendtimes % 1000 == 0) {
						try {
							Thread.sleep(100);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}
				
			});
			stream.close();
				
		}
         ReferenceCountUtil.release(sendableMsg.getMsgContent());
	}

	/**
	 * 
	 * @param sendableMsg
	 * @return 返回使用消息的次数
	 */
	public  void sendMsg2Client(SendableMsg sendableMsg) {

		String clientid=sendableMsg.getRecvclientid();
		if(str2channel.containsKey(clientid)){
			final Channel channel=
					str2channel.get(clientid);
			
			MqttQoS qos=sendableMsg.getQoS();
			int  dupTimes=sendableMsg.getDupTimes();
			
			
			 ByteBuf sendbyteBuf=sendableMsg.getMsgContent().retainedDuplicate();
			 sendbyteBuf.resetReaderIndex();
			
			MqttFixedHeader Header = new MqttFixedHeader(MqttMessageType.PUBLISH,
					dupTimes>0, qos, sendableMsg.isRetain(), 0);
			MqttPublishVariableHeader publishVariableHeader = new MqttPublishVariableHeader(
					sendableMsg.getTopname(), sendableMsg.getMessageid());
			MqttPublishMessage publishMessage = new MqttPublishMessage(Header,
					publishVariableHeader, sendbyteBuf);
			
	       
			if (qos==MqttQoS.EXACTLY_ONCE) {
				reSendService.submitDelayMsg(new DelaySnedMsg(sendableMsg));
				
			}
			
			 logger.debug(sendableMsg.getSendclientid()+"发给了:"+clientid+"主题:"+sendableMsg.getTopname());
			channel.writeAndFlush(publishMessage);

		}
		else{ //如果订阅表存在但是不在线就存入未成功队列；
			
			SendMsg msg = new SendMsg(
					sendableMsg.getRecvclientid(),
					sendableMsg.getMessageid(),
					MSG_SENDUNSUCCESS,
					sendableMsg.getSendclientid());
			    delayOCalQueue.offer(new DelayWillRemoveMsg(msg));
			    logger.debug(clientid+"不在线了，直接存在未成功队列");
			
		}
		

	}

	/**
	 * 更改发送消息状态
	 * 
	 * @param messageid 
	 * @param iden
	 */
	public  void updateSendMsg(final String clientiden, Integer msgid) {

	
		 List<SendMsg> msgs=new ArrayList<>();

		DelaySnedMsg delaySnedMsg=reSendService.getDelaySnedMsg(clientiden, msgid);
		
		if(delaySnedMsg!=null){
			msgs.add(new SendMsg(delaySnedMsg.getSendableMsg().getRecvclientid(),
		    		  msgid, MSG_SENDSUCCESS,
		    		  delaySnedMsg.getSendableMsg().getSendclientid()));
		}
		
		if(msgs==null||msgs.isEmpty()){
			
			Stream<DelayWillRemoveMsg> stream=
					delayOCalQueue.getDelayQueue().parallelStream();
			
			stream.forEach((DelayWillRemoveMsg delaymsg)->{
				SendMsg sendMsg = delaymsg.getSendMsg();
				if (sendMsg.getClientid().equals(clientiden)
						&& sendMsg.getSendmsgid() == msgid) {
					(sendMsg).setSendstate(MSG_SENDSUCCESS);
					msgs.add(sendMsg);
				}
			});
			stream.close();
		}
		
		if(msgs!=null&&msgs.isEmpty()){
			
			Stream<SendMsg> stream=msgs.parallelStream();
			stream.forEach((SendMsg sendmsg)->{
				
				if (!readUpdateSendedQue.offer(sendmsg)) {
					List<SendMsg> msgsaves = new ArrayList<SendMsg>();
					readUpdateSendedQue.drainTo(msgsaves);
					msgsaves.add(sendmsg);
					submitSendedMsg(msgsaves);

				}
			});
		}
		
	}

	
	/**
	 * 提交sendmsg
	 * 
	 * @param sedmsgs
	 */
	private  void submitSendedMsg(final List<SendMsg> sedmsgs) {

		dboptgroup.submit(()->{
			for (SendMsg sendMsg : sedmsgs) {
				sendMsgMapper.insert(sendMsg);
			}
		});

	}

	public  ConcurrentHashMap<String, Channel> getStr2channel() {
		return str2channel;
	}

	public  void setStr2channel(
			ConcurrentHashMap<String, Channel> str2channel) {
		this.str2channel = str2channel;
	}

	public  ConcurrentNavigableMap<String, BlockingQueue<String>> getSubmap() {
		return submap;
	}

	public  void setSubmap(
			ConcurrentNavigableMap<String, BlockingQueue<String>> submap) {
		this.submap = submap;
	}
	

	public  void setDboptgroup(ExecutorService dboptgroup) {
		this.dboptgroup = dboptgroup;
	}

	public ExecutorService getDboptgroup() {
		return dboptgroup;
	}

	public void setUsedb(boolean usedb) {
		this.usedb = usedb;
	}

	public BlockingQueue<SendMsg> getReadUpdateSendedQue() {
		return readUpdateSendedQue;
	}

	public void setReadUpdateSendedQue(BlockingQueue<SendMsg> readUpdateSendedQue) {
		this.readUpdateSendedQue = readUpdateSendedQue;
	}

	
	
	
     
}
